package com.gvsoft.communication.server;

import com.gvsoft.communication.console.ControlListener;
import com.gvsoft.communication.console.Console;
import com.gvsoft.communication.server.net.ChannelListener;
import com.gvsoft.communication.server.net.handle.inf.IPacketHandle;
import com.gvsoft.communication.server.net.socket.NSocket;
import com.gvsoft.communication.server.net.socket.NSocketCleanThread;
import com.gvsoft.communication.server.net.socket.NSocketManager;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created with IntelliJ IDEA.
 * ProjectName:gvMsgRouting
 * Author: zhaoqiubo
 * Date: 15/9/7
 * Time: 上午10:45
 * Desc: 消息路由启动类
 */
public class Server implements Closeable {

    private final static Logger logger = LogManager.getLogger("server");

    //目前我理解ServerSocketChannel就是SocketChannel通道的容器
    private ServerSocketChannel serverSocketChannel;
    //网络监听器的索引
    private AtomicInteger listenerIndex = new AtomicInteger(0);
    //网络监听器数组
    private ChannelListener[] listeners;
    //针对接收到的报文进行处理的线程池,报文处理执行器
    private ExecutorService readHandleExecutor;
    /**
     * 管理控制监听器
     */
    private static ControlListener controlListener;
    /**
     * 超时通道清理线程
     */
    NSocketCleanThread nSocketCleanThread;
    public static Server instance = new Server();

    private boolean isClosed = false;

//    private ExecutorService threadPool;

    public static void main(String arg[]) throws Exception {
        final Server server = Server.getInstance();
        server.start(arg);
    }

    public static Server getInstance() {
        return instance;
    }

    private Server() {
    }

    /**
     * 启动各类服务
     */

    public void start(String arg[]) {

        try {

            ServerConfig cfg = new ServerConfig(arg[0]);
            //打开一个serversocket通道，ServerSocketChannel是一个监控是否有新连接进入的通道。
            serverSocketChannel = ServerSocketChannel.open();
            //将这个serversokect通道设置为非阻塞模式
            serverSocketChannel.configureBlocking(false);
            //绑定serversokect的ip和端口
            serverSocketChannel.socket().bind(new InetSocketAddress(cfg.getIp(), cfg.getPort()));
            startThread(cfg);
            addShutdownHook();

        } catch (Exception e) {
            logger.info(e.getMessage() + "，初始化配置失败！请检查配置文件！");
        }

    }

    private void startThread(ServerConfig cfg) throws IOException {
        //启动多路复用器
        listeners = new ChannelListener[ServerConfig.getListenerCount()];
        for (int i = 0; i < ServerConfig.getListenerCount(); i++) {
            listeners[i] = new ChannelListener(this, "NO." + (i + 1) + " LISTENER");
            if (i == 0) {
                serverSocketChannel.register(listeners[i].getSelector(), SelectionKey.OP_ACCEPT);
            }
            listeners[i].start();
            logger.info(listeners[i].getName() + " Starting…………");
        }
        logger.info("通讯服务已经启动IP：" + cfg.getIp() + "，端口："
                + cfg.getPort() + "，【1】号监听器开始接收客户端连接…………");
//        threadPool = Executors.newCachedThreadPool();
        nSocketCleanThread = new NSocketCleanThread(cfg.getCleanThreadOutTime(), cfg.getCleanThreadCycle());
        nSocketCleanThread.start();
        //根据配置启动报文处理线程池
        readHandleExecutor = new ThreadPoolExecutor(ServerConfig.getReadHandleThreadCount()
                , ServerConfig.getReadHandleThreadCount(), 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(ServerConfig.getReadHandleThreadCount() * 30));
        //启动控制管理监听器
        controlListener = new ControlListener("ControlListener",cfg.getConsolePort());
        controlListener.start();

    }

    public void addShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() {
                try {
                    if(!isClosed) {
                        close();
                    }
                } catch (Exception e) {
                    logger.info("close error:" + e.getMessage());

                }
            }
        });
    }

    public void close() throws IOException {

        isClosed = true;
        readHandleExecutor.shutdownNow();
        //threadPool.shutdownNow();
        controlListener.interrupt();
        nSocketCleanThread.interrupt();
        Console.close();
        Map<String, NSocket> unregNSockets = NSocketManager.getUnregNSockets();
        for (String token : unregNSockets.keySet()) {
            NSocket nSocket = NSocketManager.getUnregNSoket(token);
            if (nSocket != null) {
                nSocket.close();
            }
            NSocketManager.removeUnregNSocket(token);
        }
        Map<String, NSocket> connCache = NSocketManager.getConnCache();
        for (String token : connCache.keySet()) {
            NSocket nSocket = NSocketManager.getConnByToken(token);
            if (nSocket != null) {
                nSocket.close();
            }
            NSocketManager.removeNSocket(nSocket.getToken(), nSocket.getUserId());
        }
        for (int i = 0; i < listeners.length; i++) {
            listeners[i].interrupt();
        }
        if (serverSocketChannel != null) {
            logger.info("GVServer消息路由服务关闭！");
            serverSocketChannel.close();
        }

    }
    public void exit(){
        System.exit(0);
    }

    /**
     * 获取新的selector分配给新连接
     */
    public ChannelListener getNextListener() {
        if (ServerConfig.getListenerCount() <= 1) {
            return listeners[0];
        } else {
            return listeners[this.listenerIndex.getAndIncrement() % ServerConfig.getListenerCount()];
        }
    }

    /**
     * 启动读取处理线程
     */
    public void execReadHandle(Runnable runnable) {
        readHandleExecutor.submit(runnable);
    }

    public void registerHandle(String cmd, IPacketHandle handle) {
        ServerConfig.PACKET_HANDLE_CLASS_MAP.put(cmd, handle);
    }

}
